#include "stdafx.h"
#include "BaseActor.h"
#include <assert.h>

//exception: generic exception that actor failed to handle a message correctly
class CBadMessageHandling: public tbb::movable_exception<void*>
{
public:
    CBadMessageHandling()
        : tbb::movable_exception<void*>(NULL)
    {}

    virtual const char* what() const throw()
    {
        return "actor failed to handle message";
    }
};


CBaseActor::CBaseActor(CBaseActor * pParent)
{
    m_pParent               = pParent;
    m_pContinuation         = NULL;
    m_pEventProcessor       = NULL;		//atomic value can't be defaulted
    if(m_pParent)
    {
        //use parent's processing unit
        m_pProcessingRoot = m_pParent->m_pProcessingRoot;
    }
    else
    {
        //become a new root processing unit/task
        m_pProcessingRoot = new(tbb::task::allocate_root())tbb::empty_task();
        m_pProcessingRoot->set_ref_count(1);
    }
    assert(m_pProcessingRoot);
    _Run();
}


CBaseActor::~CBaseActor(void)
{
    _Cancel();
}

bool CBaseActor::SendEvent( CBaseActor * target, TEvent msg )
{
    assert(target);
    target->m_messageQueue.push(msg);
    return target->_TryProcessing(m_pContinuation);
}

bool CBaseActor::SendException( CBaseActor * target, tbb::tbb_exception& e, const TEvent & msg )
{
    assert(target);
    target->m_exceptionQueue.push(std::make_pair(e.move(),msg));
    return target->_TryProcessing(m_pContinuation);
}

void CBaseActor::_Run()
{
    if(!m_pParent && m_pProcessingRoot != NULL )
    {
        tbb::spin_mutex::scoped_lock(m_destructorMutex);
        if(m_pProcessingRoot != NULL && m_pProcessingRoot->ref_count()>0)
        {
            m_pProcessingRoot->wait_for_all();
            m_pProcessingRoot->set_ref_count(1);
        }
    }
}

void CBaseActor::_Cancel()
{
    if(!m_pParent && m_pProcessingRoot != NULL )
    {
        tbb::spin_mutex::scoped_lock(m_destructorMutex);
        if(m_pProcessingRoot != NULL)
        {
            m_pProcessingRoot->cancel_group_execution();
            m_pProcessingRoot = NULL;
        }
    }
}

bool CBaseActor::_ProcessAllMessages()
{
    try
    {
        TEvent msg;
        try
        {
            bool workleft = true;
            while(workleft) //check for work at least once
            {
                workleft = false;
                TExEvent e;
                while(m_exceptionQueue.try_pop(e))
                {
                    workleft = true;
                    _PreReceive();
                    msg = e.second;
                    if(_ReceiveException(*e.first,msg))
                    {
                        e.first->destroy();//clean up handled exception
                    }
                    else
                    {
                        e.first->throw_self();
                    }
                    _PostReceive();
                }
                while(m_messageQueue.try_pop(msg))
                {
                    workleft = true;
                    _PreReceive();
                    if(!_Receive(msg) && !_ReceiveException(CBadMessageHandling(), msg))
                    {
                        throw CBadMessageHandling();
                    }
                    _PostReceive();
                }
            };
            //from this point a new task may be allocated for processing
            m_pEventProcessor = NULL;
        }
        catch(tbb::tbb_exception& e)
        {
            _PostReceive();
            if(m_pParent) //try to bubble exception to parent
            {
                SendException(m_pParent,e,msg);
                return false;
            }
            e.throw_self();
        }
        catch(std::exception& e)
        {
            _PostReceive();
            tbb::captured_exception ce(typeid(e).name(),e.what());
            if(m_pParent) //try to bubble exception to parent
            {
                SendException(m_pParent,ce,msg);
                return false;
            }
            throw;
        }
    }
    catch(...)
    {
        //root failed to handle exception so terminate root
        return false;
    }
    return true;
}

bool CBaseActor::_TryProcessing( tbb::empty_task* continuation )
{
    //NOTE: if NULL set a sentinel as placeholder
    if(m_pEventProcessor.compare_and_swap((CEventProcessor *)(0x1),NULL)==NULL)
    {
        if(!continuation)
        {
            continuation = new( m_pProcessingRoot->allocate_continuation() ) tbb::empty_task;
        }
        assert(continuation);
        m_pContinuation = continuation;

        m_pEventProcessor = new(continuation->allocate_child()) CEventProcessor(this);
        assert(m_pEventProcessor);

        continuation->increment_ref_count();
        continuation->spawn(*m_pEventProcessor);
        //_Run();
    }
    return true;
}


//////////////////////////////////////////////////////////////////////////
tbb::task* CBaseActor::CEventProcessor::execute()
{
    bool ok = m_pActor->_ProcessAllMessages();
    m_pActor->m_pContinuation = NULL;
    m_pActor->m_pEventProcessor = NULL;
    //if(!ok)
    //{
    //    m_pActor->_Destroy();
    //};
    return NULL;
}
